package netty.task;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.github.benmanes.caffeine.cache.Cache;
import com.github.benmanes.caffeine.cache.Caffeine;
import netty.session.NettySession;
import netty.session.SessionManager;
import netty.task.invoker.RemoteCmdInvoker;
import org.springframework.data.redis.core.StringRedisTemplate;
import redis.task.DeferredResultTask;
import redis.task.DeferredResultTaskManager;
import redis.task.DeferredResultTaskResult;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

public class NettyDeferredTaskManager {
    private Cache<String, DeferredResultTask> taskMap; // key为sessionId + taskType
    private ObjectMapper objectMapper;
    private SessionManager sessionManager;
    private StringRedisTemplate redisTemplate;
    private List<RemoteCmdInvoker> invokers;

    public NettyDeferredTaskManager(SessionManager sessionManager, StringRedisTemplate redisTemplate) {
        this.taskMap = Caffeine.newBuilder().expireAfterWrite(1, TimeUnit.MINUTES).build();
        this.objectMapper = new ObjectMapper();
        this.sessionManager = sessionManager;
        this.redisTemplate = redisTemplate;
        this.invokers = new ArrayList<>();
    }

    public void execute(DeferredResultTask<?> task) {
        NettySession session = sessionManager.getSession(task.getSessionId());
        try {
            if (session == null) {
                    redisTemplate.convertAndSend(DeferredResultTaskManager.getResultChannel(task.getFromServerId()),
                            objectMapper.writeValueAsString(DeferredResultTaskResult.offline(task)));
            } else {
                if (taskMap.getIfPresent(task.getSessionId() + task.getType()) != null) {
                    return; // 重复任务，直接返回
                }
                RemoteCmd cmd = RemoteCmd.fromTask(task, session.getChannel());
                RemoteCmdInvoker invoker = invokers.stream().filter(i -> i.support(cmd)).findFirst().orElse(null);
                if (invoker != null) {
                    taskMap.put(task.getSessionId() + task.getType(), task); // 添加到taskMap，Netty收到响应数据后向redis发送结果，同时从taskMap移除任务
                    invoker.invoke(cmd);
                } else {
                    redisTemplate.convertAndSend(DeferredResultTaskManager.getResultChannel(task.getFromServerId()),
                            objectMapper.writeValueAsString(DeferredResultTaskResult.notSupport(task)));
                }
            }
        } catch (JsonProcessingException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 收到任务结果后，向对方服务器发送消息
     * @param sessionId
     * @param taskType
     * @param data
     * @param <T>
     */
    @SuppressWarnings("unchecked")
    public <T> void receivedResult(String sessionId, String taskType, T data) {
        DeferredResultTask task = taskMap.getIfPresent(sessionId + taskType);
        if (task != null) {
            taskMap.invalidate(sessionId + taskType);
            DeferredResultTaskResult<T> result = DeferredResultTaskResult.succeed(task, data);
            try {
                redisTemplate.convertAndSend(DeferredResultTaskManager.getResultChannel(task.getFromServerId()),
                        objectMapper.writeValueAsString(result));
            } catch (JsonProcessingException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public void addCmdInvoker(RemoteCmdInvoker invoker) {
        invokers.add(invoker);
    }

    public List<RemoteCmdInvoker> getInvokers() {
        return invokers;
    }

    public void setInvokers(List<RemoteCmdInvoker> invokers) {
        this.invokers = invokers;
    }
}
